package day07

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

object TypedDemo {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession.builder().appName("TypedDemo").master("local[2]").getOrCreate()
    import spark.implicits._

    val employee: DataFrame = spark.read.json("E://data/employee.json")
    //将DataFrame转换为RDd类型
    val employeeRdd: RDD[Row] = employee.rdd
    println("初始分区数为：" + employeeRdd.partitions.size)

    //1.重分区，注意DataFrame和RDD这两个类型都可以重分区，这里将DataFrame重分区。
    val empRep: Dataset[Row] = employee.repartition(8)
    val empCoa: Dataset[Row] = employee.coalesce(7) //默认shuffle为false
    println("repartition: "+empRep.rdd.partitions.size) //返回8
    println("coalese: "+empCoa.rdd.partitions.size)     //返回1

    //2. distinct、dropDuplicates。两者都是去重。
    // distinct是按照整行是否相同来进行去重，
    // dorpDuplicates(里面的参数可以为Array("字段"))是按照指定的一个或者多个字段进行比对去重，发现有重复的只取第一条。若不加参数则与dinstinc一样
    employee.distinct().show()
    employee.dropDuplicates(Array("name")).show()

    //3. expect、filter
    val employee2: DataFrame = spark.read.json("E://data/employee2.json")
    /*
    employee.filter(emp => emp.name)
    注意，上面这个语句如果employee是DF类型的话，在emp.name会报错，获取不到这个name字段数据
    因为filter方法是强类型Tyoed的，因此employee必须得是DataSet。
     */
    // 获取当前DataSet中有但是另外一个DataSet中没有的元素，相当于差集
    val employeeDS: Dataset[Employee] = employee.as[Employee]
    val employee2DS: Dataset[Employee] = employee2.as[Employee]

/*
    println("----------except--------------------")
    employeeDS.except(employee2DS).show()   //参数：other: Dataset[Employee]
    employeeDS.filter(emp => emp.age>30).show() //过滤，满足就保留，否则就过滤掉
    println("----------union--------------------")
    employeeDS.union(employee2DS).show()      //并集，不会去重
    println("----------intersect--------------------")
    employeeDS.intersect(employee2DS).show()  //交集
*/
    // joinWith 、 sort
    val departmentDS: Dataset[Department] = spark.read.json("E://data/department.json").as[Department]
    println("----------joinWith--------------------")
    employeeDS.joinWith(departmentDS, $"depId"===$"id").show()
    employeeDS.sort($"age".desc).show()   //desc是降序

    //collect_set：将指定字段的值都收集到一起，会按照这个字段进行去重
    //collect_list同上，但是不会按照字段去重
    import org.apache.spark.sql.functions._
    println("----------collect_set 和 collect_list------------------")
    employeeDS
      .groupBy(employeeDS("depId"))
      .agg(collect_set(employeeDS("name")), collect_list(employeeDS("name")))
      .show()

    // avg、sum、max、min、count、countDistinct
    // 这些DataFrame类型都可以做，因为都是untyped
    employeeDS
        .join(departmentDS,$"depId"===$"id")
        .groupBy(departmentDS("name"))
        .agg(avg(employeeDS("salary")),  sum(employeeDS("salary")),
          max(employeeDS("salary")), min(employeeDS("salary")),
          count(employeeDS("salary")), countDistinct(employeeDS("salary")))
      //countDistinct(employeeDS("salary"))是统计salary不同的个数，与其他字段无关
        .show()

    // untyped：select where join groupBy agg

    spark.stop()

  }
}
//因为salary有小数，所以为Double类型
case class Employee(name:String, age:Long, depId:Long, gender:String, salary:Double)

case class Department(id:Long, name:String)
